package com.bawei.gmall.realtime.app

import com.alibaba.fastjson.JSON
import com.bawei.gmall.common.GmallConstants
import com.bawei.gmall.realtime.bean.{OrderDetail, OrderInfo, SaleDetail, UserInfo}
import com.bawei.gmall.realtime.util.{EsUtil, KafkaUtil}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.json4s.jackson.Serialization
import redis.clients.jedis.Jedis

import scala.collection.mutable.ListBuffer

object SaleApp {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("sale_app").setMaster("local[*]")
    val ssc = new StreamingContext(conf, Seconds(5))
    // 1. 消费kafka中的数据
    val orderRecordDStream: InputDStream[ConsumerRecord[String, String]] = KafkaUtil.getKafkaStream(GmallConstants.KAFKA_TOPIC_ORDER, ssc)
    val orderDetailRecordDStream: InputDStream[ConsumerRecord[String, String]] = KafkaUtil.getKafkaStream(GmallConstants.KAFKA_TOPIC_ORDER_DETAIL, ssc)
    val userRecordDStream: InputDStream[ConsumerRecord[String, String]] = KafkaUtil.getKafkaStream(GmallConstants.KAFKA_TOPIC_USER_INFO, ssc)

    // 2. 转换结构 -> orderInfo
    val orderDStream: DStream[OrderInfo] = orderRecordDStream.map {
      record => {
        val orderInfo = JSON.parseObject(record.value(), classOf[OrderInfo])
        // 补充时间字段
        val dataTimes = orderInfo.create_time.split(" ")
        orderInfo.create_date = dataTimes(0)
        orderInfo.create_hour = dataTimes(1).split(":")(0)
        // 脱敏
        val tuple = orderInfo.consignee_tel.splitAt(4)
        orderInfo.consignee_tel = tuple._1 + "*******"
        println("主表！" + orderInfo)
        orderInfo
      }
    }
    val orderInfoWithKeyDStream = orderDStream.map(orderInfo => (orderInfo.id, orderInfo))

    // 转换结构 -> orderDetail
    val orderDetailDStream = orderDetailRecordDStream.map {
      record => {
        val orderDetail = JSON.parseObject(record.value(), classOf[OrderDetail])
        println("从表！" + orderDetail)
        orderDetail
      }
    }
    val orderDetailWithKeyDStream = orderDetailDStream.map(orderDetail => (orderDetail.order_id, orderDetail))
    // orderInfo join orderDetail
    val fullJoinDStream: DStream[(String, (Option[OrderInfo], Option[OrderDetail]))] = orderInfoWithKeyDStream.fullOuterJoin(orderDetailWithKeyDStream)
    // 如果orderInfo不为空
    // 1. 如果从表也不为空直接关联
    // 2. 把自己写入缓存
    // 3. 查询缓存
    val saleDetailDStream: DStream[SaleDetail] = fullJoinDStream.flatMap {
      case (orderId, (orderInfoOpt, orderDetailOpt)) => {
        val saleDetailList = ListBuffer[SaleDetail]()
        val jedis = new Jedis("hadoop102", 6379)
        // 引入Json4s
        implicit val formats = org.json4s.DefaultFormats
        // 如果主表不为空
        if (orderInfoOpt != None) {
          val orderInfo = orderInfoOpt.get
          // 从表不为空
          if (orderDetailOpt != None) {
            val orderDetail = orderDetailOpt.get
            val saleDetail = new SaleDetail(orderInfo, orderDetail)
            saleDetailList += saleDetail
          }
          // 从表为空，主表放入redis
          val orderInfoKey = "order_info:" + orderInfo.id
          val orderInfoJson = Serialization.write(orderInfo)
          jedis.setex(orderInfoKey, 3600, orderInfoJson)
          // 查缓存获取orderDetail信息
          val orderDetailKey = "order_detail:" + orderInfo.id
          val orderDetailSet = jedis.smembers(orderDetailKey)
          // 引入Java支持
          import scala.collection.JavaConversions._
          for (orderDetailJson <- orderDetailSet) {
            val orderDetail = JSON.parseObject(orderDetailJson, classOf[OrderDetail])
            val saleDetail: SaleDetail = new SaleDetail(orderInfo, orderDetail)
            saleDetailList += saleDetail
          }
        }
        // 如果主表为空，从表不为空
        else if (orderDetailOpt != None) {
          // 1. 把自己写入缓存
          val orderDetail = orderDetailOpt.get
          val orderDetailJson = Serialization.write(orderDetail)
          val orderDetailKey = "order_detail:" + orderDetail.order_id
          jedis.sadd(orderDetailKey, orderDetailJson)
          jedis.expire(orderDetailKey, 3600)
          // 2. 查缓存
          val orderInfoKey = "order_info:" + orderDetail.order_id
          val orderInfoJson = jedis.get(orderInfoKey)
          if (orderInfoJson != null && orderDetailJson.size > 0) {
            val orderInfo = JSON.parseObject(orderInfoJson, classOf[OrderInfo])
            val saleDetail = new SaleDetail(orderInfo, orderDetail)
            saleDetailList += saleDetail
          }
        }
        jedis.close()
        saleDetailList
      }
    }

    // saleDetailDStream join userInfo
    val fullSaleDetailDStream: DStream[SaleDetail] = saleDetailDStream.mapPartitions {
      saleDetailItr => {
        val jedis = new Jedis("hadoop102", 6379)
        val saleDetailList = new ListBuffer[SaleDetail]
        for (saleDetail <- saleDetailItr) {
          val userInfoJson = jedis.get("user_info:" + saleDetail.user_id)
          if (userInfoJson != null) {
            val userInfo = JSON.parseObject(userInfoJson, classOf[UserInfo])
            saleDetail.mergeUserInfo(userInfo)
          }
          saleDetailList += saleDetail
        }
        jedis.close()
        saleDetailList.toIterator
      }
    }

    // 存入ES
    fullSaleDetailDStream.foreachRDD {
      rdd => {
        rdd.foreachPartition {
          saleDetailItr => {
            val dataList: List[(String, SaleDetail)] = saleDetailItr.map(saleDetail => (saleDetail.order_detail_id, saleDetail)).toList
            EsUtil.indexBulk(GmallConstants.ES_INDEX_SALE, dataList)
          }
        }
      }
    }

    // user新增变化数据写入缓存
    userRecordDStream.map(record => JSON.parseObject(record.value(), classOf[UserInfo])).foreachRDD {
      rdd => {
        rdd.foreachPartition {
          userInfoItr => {
            val jedis = new Jedis("hadoop102", 6379)
            implicit val formats = org.json4s.DefaultFormats
            for (elem <- userInfoItr) {
              val userInfoJson = Serialization.write(elem)
              val userInfoKey = "user_info:" + elem.id
              println("userInfo!" + elem)
              jedis.set(userInfoKey, userInfoJson)
            }
            jedis.close()
          }
        }
      }
    }

    ssc.start()
    ssc.awaitTermination()
  }
}
